package com.zimmor.mq.rocket_train.base;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Author:zimmor
 * @Date: 2023/2/14 16:26
 * @Description pull方式消费消息
 * @version: 1.0.0
 */

public class PullDefaultConsumer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //创建Pull消费者
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consumer-group5");
        //设置namesrv
        consumer.setNamesrvAddr("192.168.204.136:9876;192.168.204.137:9876");
        //启动consumer
        consumer.start();
        //构建MessageQueue
        MessageQueue messageQueue = new MessageQueue();
        messageQueue.setBrokerName("broker-a");
        messageQueue.setQueueId(0);
        messageQueue.setTopic("base1");
        long offset = 40;
        //拉取消息
        for (int i = 0; i <10 ; i++) {

            PullResult pullResult = consumer.pull(messageQueue, "tag1", offset, 10);
            //解析拉取结果
            if (PullStatus.FOUND.equals(pullResult.getPullStatus())) {
                List<MessageExt> msgFoundList = pullResult.getMsgFoundList();
                for (MessageExt messageExt : msgFoundList) {
                    System.out.println("msg = " + new String(messageExt.getBody()) + "messageID" + messageExt.getMsgId());
                }
//                consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset());
                offset =  pullResult.getNextBeginOffset();
                TimeUnit.SECONDS.sleep(10);
                System.out.println("========================");
            }
        }
        //关闭consumer
        consumer.shutdown();
    }
}
